/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl.kryo;

import org.sabayframework.channel.Channel;
import org.sabayframework.channel.impl.NullChannel;
import org.sabayframework.channel.ram.ConcurrentArrayChannel;
import org.sabayframework.cluster.ComputerId;
import org.sabayframework.log.Log;
import org.sabayframework.queue.ExecutionQueue;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryonet.Connection;
import com.esotericsoftware.kryonet.Listener;

public class KryoListener extends Listener{
	
	private final RequestManager requestManager;
	private NetworkManager networkManager;
	private final ConcurrentArrayChannel<Request> requestQueue;
	
	public KryoListener(RequestManager requestManager,
			ExecutionQueue executionQueue,
			Kryo kryo) {
		super();
		this.requestManager = requestManager;
		this.requestQueue = new ConcurrentArrayChannel<Request>(100, true);
		RequestActivity a = new RequestActivity(requestManager, kryo);
		final Channel<Request> nullOut = NullChannel.getInstance(); 
		executionQueue.submit(a, requestQueue, nullOut);
	}

	public void setNetworkManager(NetworkManager networkManager) {
		this.networkManager = networkManager;
	}

	@Override
	public void disconnected(Connection connection) {
		networkManager.removeConnection(connection.getID());
	}

	@Override
	public void received(Connection connection, Object object) {
		if(!(object instanceof NetworkPacket)) return;
		NetworkPacket packet =(NetworkPacket) object;
		networkManager.addConnection(packet.getRemoteId(), connection.getID());
		processNetworkPacket(packet);
	}
	
	public void processNetworkPacket(NetworkPacket packet){
		Log.debug("KryoListener.process ", packet);
		ComputerId remoteId = packet.getRemoteId();
		DistributedActivityContext dac = null;
		switch(packet.getType()){
		case START:
			dac = requestManager.addLocalActivity(remoteId, packet); break;
		case INPUT:
			dac = requestManager.findLocalActivity(remoteId, packet);
			if(dac == null && packet.getActivity()!=null) {
				//first time request
				dac = requestManager.addLocalActivity(remoteId, packet);
			}
			break;
		case CLOSE_INPUT:
			dac = requestManager.findLocalActivity(remoteId, packet);
			if(dac == null) return;
			break;
		case OUTPUT:
			dac = requestManager.findRemoteActivity(packet); break;
		case CLOSE_OUTPUT:
			dac = requestManager.findRemoteActivity(packet); 
			if(dac == null) return;
			break;
		}
		if(dac == null) {
			//TODO debug error handling
			final int remoteActivityId = packet.getActivityId();
			dac = new DistributedActivityContext(remoteActivityId);
			dac.setClientId(remoteId);
			Throwable error = new IllegalArgumentException("Activity not found "+remoteActivityId);
			Request r = new Request(RequestType.CLOSE_OUTPUT, dac, error);
			requestManager.submitRequest(r);
		}
		else {
			if(packet.getActivity()!=null){
				Request r = new Request(RequestType.START, dac, packet.getActivity());
				submitRequest(r, dac);
			}
			Request r = new Request(packet.getType(), dac, packet.getData());
			submitRequest(r, dac);
		}
	}
	private void submitRequest(Request r, DistributedActivityContext dac){
		try{
			requestQueue.put(r);
		} catch (Exception e) {
			r = new Request(RequestType.CLOSE_OUTPUT, dac, e);
			requestManager.submitRequest(r);
		}
		
	}
}
